/*
 * Copyright (c) 2012-2014 The original author or authors
 * ------------------------------------------------------
 * All rights reserved. This program and the accompanying materials
 * are made available under the terms of the Eclipse Public License v1.0
 * and Apache License v2.0 which accompanies this distribution.
 *
 * The Eclipse Public License is available at
 * http://www.eclipse.org/legal/epl-v10.html
 *
 * The Apache License v2.0 is available at
 * http://www.opensource.org/licenses/apache2.0.php
 *
 * You may elect to redistribute this code under either of these licenses.
 */
package org.dna.mqtt.moquette.messaging.spi.impl;

import java.util.Properties;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

import org.dna.mqtt.moquette.messaging.spi.IMessaging;
import org.dna.mqtt.moquette.messaging.spi.impl.events.DisconnectEvent;
import org.dna.mqtt.moquette.messaging.spi.impl.events.InitEvent;
import org.dna.mqtt.moquette.messaging.spi.impl.events.LostConnectionEvent;
import org.dna.mqtt.moquette.messaging.spi.impl.events.MessagingEvent;
import org.dna.mqtt.moquette.messaging.spi.impl.events.ProtocolEvent;
import org.dna.mqtt.moquette.messaging.spi.impl.events.StopEvent;
import org.dna.mqtt.moquette.processor.impl.MqttProtocolProcessor;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import org.dna.mqtt.moquette.server.ServerChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.lmax.disruptor.BatchEventProcessor;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.SequenceBarrier;

/**
 * {@code IMessaging} implementation that handles all the messaging input from Netty I/O handler. 
 * It simply publishes all the inbound message to a ring buffer for further event-based asynchronous 
 * processing by a {@code EventHandler<ValueEvent>} implementation {@code MqttProtocolEventHandler}. 
 * 
 * @author puran, andrea
 */
public class MessageInputDisruptor implements IMessaging {
	private static final Logger LOG = LoggerFactory.getLogger(MessageInputDisruptor.class);
	
	/* LMAX disruptor singleton instance, there should be one and only one inbound disruptor instance per broker instance */
	private static MessageInputDisruptor INPUT_DISRUPTOR_INSTANCE;
	
	/* LMAX disruptor related variables */
	private RingBuffer<ValueEvent> m_ringBuffer;
	private BatchEventProcessor<ValueEvent> m_eventProcessor;
	private ExecutorService m_executor;
	private CountDownLatch 	m_stopLatch;
	
	/* Protocol event handler also LMAX event handler implementation */
	private MqttProtocolProcessor m_protocolEventHandler;
	
	/**
	 * Returns singleton inbound disruptor instance.
	 * @return the disruptor instance
	 */
	public static MessageInputDisruptor getInstance() {
		if (INPUT_DISRUPTOR_INSTANCE == null) {
			INPUT_DISRUPTOR_INSTANCE = new MessageInputDisruptor();
		}
		return INPUT_DISRUPTOR_INSTANCE;
	}
	
	/**
	 * Initialize inbound disruptor, this is during the system startup.
	 * @param configProps the system config properties
	 */
	public void init(Properties configProps) {
		m_stopLatch = new CountDownLatch(1);
		m_protocolEventHandler = new MqttProtocolProcessor(m_stopLatch);
		
		m_executor = Executors.newFixedThreadPool(1);
		m_ringBuffer = new RingBuffer<ValueEvent>(ValueEvent.EVENT_FACTORY, 1024 * 32);
		
		SequenceBarrier barrier = m_ringBuffer.newBarrier();
		m_eventProcessor = new BatchEventProcessor<ValueEvent>(m_ringBuffer, barrier, m_protocolEventHandler);
		
		//TODO: in a presentation is said to don't do the followinf line!!
		m_ringBuffer.setGatingSequences(m_eventProcessor.getSequence());
		m_executor.submit(m_eventProcessor);

		disruptorPublish(new InitEvent(configProps));
	}
	
	@Override
	public void handleProtocolMessage(ServerChannel session, AbstractMessage msg) {
		disruptorPublish(new ProtocolEvent(session, msg));
	}
	
	@Override
	public void stop() {
		disruptorPublish(new StopEvent());
		try {
			// wait the callback notification from the protocol processor thread
			boolean elapsed = !m_stopLatch.await(10, TimeUnit.SECONDS);
			if (elapsed) {
				LOG.error("Can't stop the server in 10 seconds");
			}
			m_executor.shutdown();
		} catch (InterruptedException ex) {
			LOG.error("Disruptor shutdown interrupted", ex);
		}
	}

	@Override
	public void disconnect(ServerChannel session) {
		disruptorPublish(new DisconnectEvent(session));
	}

	@Override
	public void lostConnection(String clientID) {
		disruptorPublish(new LostConnectionEvent(clientID));
	}
	
	/**
	 * Publish the given message to the input disruptor ring buffer for
	 * processing.
	 * @param msgEvent the message event
	 */
	private void disruptorPublish(MessagingEvent msgEvent) {
		LOG.debug("disruptorPublish publishing event {}", msgEvent);
		long sequence = m_ringBuffer.next();
		ValueEvent event = m_ringBuffer.get(sequence);
		event.setEvent(msgEvent);

		m_ringBuffer.publish(sequence);
	}
}
